#include "config.h"

#include <sys/types.h>
#include <errno.h>

#define DBG_SUBSYS S_LIBCLUSTER

#include "net_global.h"
#include "cluster.h"
#include "dispatch.h"
#include "ynet_rpc.h"
#include "conn.h"
#include "job_dock.h"
#include "dbg.h"

static int __dispatch_newdisk__(const char *pool, diskid_t *diskid, int repnum,
                              const nid_t *skip, int skip_count, int flag)
{
        int ret;
        nid_t master;

        master = *net_getadmin();

        if (net_islocalcall(&master)) {
                ret = dispatch_srv_newdisk(pool, diskid, repnum, skip, skip_count, flag);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        } else {
                ret = dispatch_rpc_newdisk(&master, pool, diskid, repnum, skip, skip_count, flag);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int dispatch_newdisk2(const char *pool, nid_t *_nid, const nid_t *skip, int skip_count)
{
        int ret;
        uint32_t count = 2;
        nid_t nid[LICH_REPLICA_MAX];

retry:
        ret = __dispatch_newdisk__(pool, nid, count, skip, skip_count, 0);
        if (unlikely(ret)) {
                if (ret == ENOSPC) {
                        if (count ==  2) {
                                count = 1;
                                goto retry;
                        } else {
                                DBUG("no space, need %u\n", count);
                                GOTO(err_ret, ret);
                        }
                } else
                        GOTO(err_ret, ret);
        }

        if (count == 2) {
                netable_select(nid, 2, _nid);
                int s = nid_cmp(_nid, &nid[0]) == 0 ? 1 : 0;
                DBUG("skip %s\n", network_rname(&nid[s]));
        } else {
                DBUG("balance fail\n");
                *_nid = nid[0];
        }

        return 0;
err_ret:
        return ret;
}

/**
 * @param _reploc OUT
 * @param _count OUT
 * @param site_name
 * @param inrack
 * @param skip
 * @param skip_count
 * @param balance
 * @return
 */

static int __dispatch_newdisk(nid_t *_nid, int *_repnum, int repmin, const char *pool,
                              const nid_t *skip, int skip_count)
{
        int ret, count, online, total;
        uint32_t repnum = *_repnum;

        YASSERT(repnum >= repmin);

        ret = conn_faultdomain(&total, &online);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (repnum > online) {
                if (repmin <= online) {
                        count = online;
                } else {
                        ret = ENOSPC;
                        GOTO(err_ret, ret);
                }
        } else {
                count = repnum;
        }

retry:
        ret = __dispatch_newdisk__(pool, _nid, count, skip, skip_count, 0);
        if (unlikely(ret)) {
                if (ret == ENOSPC) {
                        if (count > repmin) {
                                count = repmin;
                                goto retry;
                        } else {
                                DBUG("no space, need %u\n", count);
                                GOTO(err_ret, ret);
                        }
                } else
                        GOTO(err_ret, ret);
        }

        YASSERT(count >= repmin);
        *_repnum = count;
        
        return 0;
err_ret:
        return ret;
}

int dispatch_newdisk(nid_t *_nid, int *_repnum, int repmin, const char *pool,
                     const nid_t *skip, int skip_count, int flag)
{
        int ret, repnum = *_repnum;
        nid_t nid[LICH_REPLICA_MAX];

        if ((flag & __NEWDISK_BALANCE__) && ng.daemon) {
                repnum = repnum + 1;
        }

        ret = __dispatch_newdisk(nid, &repnum, repmin, pool, skip, skip_count);
        if (unlikely(ret))
                GOTO(err_ret, ret);
        
        if (repnum > *_repnum) {
                netable_sort(nid, repnum);
                repnum = repnum - 1;
        }

        memcpy(_nid, nid, sizeof(*_nid) * repnum);
        *_repnum = repnum;
        
        return 0;
err_ret:
        return ret;
}

int dispatch_writeable(const nid_t *diskid, int count)
{
        nid_t master;

        master = *net_getadmin();

        if (net_islocalcall(&master)) {
                return dispatch_srv_writeable(diskid, count);
        } else {
                return dispatch_rpc_writeable(&master, diskid, count);
        }
}

int dispatch_newid(chkid_t *chkid)
{
        nid_t master;

        master = *net_getadmin();

        if (net_islocalcall(&master)) {
                return dispatch_srv_newid(chkid);
        } else {
                return dispatch_rpc_newid(&master, chkid);
        }
}

int dispatch_netinfo(ynet_net_info_t *info, const nid_t *nid)
{
        nid_t master;

        YASSERT(!net_isnull(nid));

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_netinfo(info, nid);
        } else {
                return dispatch_rpc_netinfo(&master, info, nid);
        }
}

int dispatch_heartbeat(const char *name, const char *dfinfo, const nodestat_t *stat,
                       const char *status, uint32_t *admin_uptime)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_heartbeat(name, dfinfo, stat, status, admin_uptime);
        } else {
                return dispatch_rpc_heartbeat(&master, name, dfinfo, stat, status, admin_uptime);
        }
}

int dispatch_vol_notifiction(const char *name, const local_vol_t *vols, uint32_t options)
{
        (void) options;
        (void) vols;
        (void) name;

        return 0;

#if ENABLE_BALANCE
        nid_t master;
        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_vol_notifiction(name, vols, options);
        } else {
                return dispatch_rpc_vol_notifiction(&master, name, vols, options);
        }
#endif

}

int dispatch_sysstat(lich_stat_t *stat, int force)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_sysstat(stat, force);
        } else {
                return dispatch_rpc_sysstat(&master, stat, force);
        }
}

int dispatch_pooldump(const char *pool)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_pooldump(pool);
        } else {
                return dispatch_rpc_pooldump(&master, pool);
        }
}

int dispatch_addnode(const char *name, const nid_t *nid)
{
        nid_t master;

        UNIMPLEMENTED(__DUMP__);

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_addnode(name, nid);
        } else {
                return dispatch_rpc_addnode(&master, name, nid);
        }
}

int dispatch_delnode(const char *name)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_delnode(name);
        } else {
                return dispatch_rpc_delnode(&master, name);
        }
}

int dispatch_check_storage_area(const char *site_name)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_check_storage_area(site_name);
        } else {
                return dispatch_rpc_check_storage_area(&master, site_name);
        }
}

int dispatch_list_storage_area(char *buf, int *count)
{
        nid_t master;

        master = *net_getadmin();
        if (net_islocalcall(&master)) {
                return dispatch_srv_list_storage_area(buf, count);
        } else {
                return dispatch_rpc_list_storage_area(&master, buf, count);
        }
}

int dispatch_init()
{
        int ret;

        ret = dispatch_srv_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = dispatch_rpc_init();
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

void dispatch_destroy()
{
        dispatch_rpc_destroy();
        dispatch_srv_destroy();
}
